package com.etl.dataflow.scheduler.communication;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.etl.dataflow.common.collection.queue.PriorityQueue;
import com.etl.dataflow.common.collection.queue.PriorityQueueImpl;
import com.etl.dataflow.common.entity.ExecutorInfo;
import com.etl.dataflow.common.entity.TaskInfo;
import com.etl.dataflow.common.enums.TaskStateEnum;
import com.etl.dataflow.common.util.ResponseUtil;
import com.etl.dataflow.scheduler.admin.entity.DataflowExecutorInfo;
import com.etl.dataflow.scheduler.admin.entity.DataflowTaskRecord;
import com.etl.dataflow.scheduler.admin.entity.vo.DataflowExecutorInfoVo;
import com.etl.dataflow.scheduler.admin.mapper.DataflowExecutorInfoMapper;
import com.etl.dataflow.scheduler.admin.mapper.DataflowTaskRecordMapper;
import com.etl.dataflow.scheduler.admin.service.ExecutorInfoService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

@Slf4j
@Component
public class ExecutorStateChecker {

    private final Map<String, ExecutorInfo> execInfoMap = new ConcurrentHashMap<>();

    private final ExecutorService pollWorkers = Executors.newFixedThreadPool(4);

    private final AtomicBoolean pollStarted = new AtomicBoolean(false);
    private final AtomicBoolean stop = new AtomicBoolean(false);

    private static final Lock EXECUTOR_OFFLINE_LOCK = new ReentrantLock();
    private static final Condition EXECUTOR_OFFLINE_CONDITION = EXECUTOR_OFFLINE_LOCK.newCondition();

    private final PriorityQueue<DataflowExecutorInfoVo> priorityQueue = new PriorityQueueImpl<>();

    @Resource
    private DataflowExecutorInfoMapper executorInfoMapper;

    @Resource
    private ExecutorInfoService executorInfoService;

    @Resource
    private DataflowTaskRecordMapper taskRecordMapper;

    private ExecutorStateChecker() {
    }

    public void register(ExecutorInfo executorInfo) {
        if (stop.get()) {
            return;
        }
        synchronized (execInfoMap) {
            execInfoMap.put(executorInfo.getUuid(), executorInfo);
            log.debug(">>>>>>>>>>>>>>执行器已经上线：" + executorInfo);
            // 通知等待执行器上线的线程恢复运行
            notifyWaiter();
        }
    }

    /**
     * 通知等待执行器上线的线程恢复运行
     */
    private void notifyWaiter() {
        EXECUTOR_OFFLINE_LOCK.lock();
        try {
            EXECUTOR_OFFLINE_CONDITION.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            EXECUTOR_OFFLINE_LOCK.unlock();
        }
    }

    public Collection<ExecutorInfo> getOnlineExecutors() {
        return this.execInfoMap.values();
    }

    public Optional<ExecutorInfo> getOptimalExecutor() {
        // 从优先队列中选取资源占用率最低的执行器
        DataflowExecutorInfoVo minimum = priorityQueue.getFront();
        if (minimum == null) {
            return Optional.empty();
        }
        if (!execInfoMap.containsKey(minimum.getUuid())) {
            return Optional.empty();
        }
        ExecutorInfo executorInfo = new ExecutorInfo();
        BeanUtil.copyProperties(minimum, executorInfo);
        return Optional.of(executorInfo);
    }

    public void stopRefreshExecState() {
        stop.set(true);
        pollWorkers.shutdown();
    }

    public void startAutoRefreshExecState() {
        // 已经开始轮询，直接返回
        if (pollStarted.get()) {
            return;
        }
        log.debug(">>>>>>>>>>>>>>>开始轮询执行器状态");
        List<DataflowExecutorInfo> executorInfoList = executorInfoMapper.selectList(null);
        if (CollUtil.isNotEmpty(executorInfoList)) {
            Map<String, ExecutorInfo> collect = executorInfoList.parallelStream()
                    .map(it -> {
                        ExecutorInfo executorInfo = new ExecutorInfo();
                        BeanUtil.copyProperties(it, executorInfo);
                        reloadTaskListByNetwork(executorInfo);
                        return executorInfo;
                    }).collect(Collectors.toMap(ExecutorInfo::getUuid, v -> v));
            execInfoMap.putAll(collect);
        }
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        executorService.execute(() -> {
            pollStarted.set(true);
            for (; ; ) {
                while (execInfoMap.isEmpty()) {
                    if (stop.get()) {
                        pollStarted.set(false);
                        return;
                    }
                    try {
                        TimeUnit.SECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                synchronized (execInfoMap) {
                    if (!execInfoMap.isEmpty()) {
                        final Collection<ExecutorInfo> values = execInfoMap.values();
                        Iterator<ExecutorInfo> iterator = values.iterator();
                        while (iterator.hasNext()) {
                            final ExecutorInfo executorInfo = iterator.next();
                            pollWorkers.execute(() -> {
                                boolean executorOnline = true;
                                HttpResponse httpResponse = null;
                                try {
                                    String pingUrl = this.getPingUrl(executorInfo);
                                    httpResponse = HttpUtil.createGet(pingUrl).setConnectionTimeout(5000).execute();
                                    String body = httpResponse.body();
                                    if (ResponseUtil.isFailed(body)) {
                                        executorOnline = false;
                                    }
                                } catch (Exception e) {
                                    executorOnline = false;
                                } finally {
                                    if (httpResponse != null) {
                                        httpResponse.close();
                                    }
                                }
                                if (executorOnline) {
                                    updateExecutorRuntimeInfo(executorInfo);
                                    reloadTaskListByNetwork(executorInfo);
                                } else {
                                    iterator.remove();
                                    executorInfoMapper.deleteById(executorInfo.getUuid());
                                    log.debug(">>>>>>>>>>>>>>执行器下线: " + executorInfo);
                                    removeExecutorRuntimeInfo(executorInfo);
                                    updateTaskRecordsToError(executorInfo);
                                }
                            });
                        }
                    }
                }
                if (stop.get()) {
                    pollStarted.set(false);
                    return;
                }
                // 进行一轮轮询后休息5s再继续
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        executorService.shutdown();
    }

    private void updateTaskRecordsToError(ExecutorInfo executorInfo) {
        List<String> taskRecordIds = new ArrayList<>();
        for (TaskInfo taskInfo : executorInfo.getTaskList()) {
            String taskRecordId = taskInfo.getTaskRecordId();
            if (taskRecordId != null) {
                taskRecordIds.add(taskRecordId);
            }
        }
        if (CollUtil.isNotEmpty(taskRecordIds)) {
            UpdateWrapper<DataflowTaskRecord> uw = new UpdateWrapper<DataflowTaskRecord>().in("id", taskRecordIds);
            DataflowTaskRecord entity = new DataflowTaskRecord()
                    .setTaskState(TaskStateEnum.FATAL_ERROR.getValue())
                    .setErrorLog("执行器下线");
            taskRecordMapper.update(entity, uw);
        }
    }

    private void removeExecutorRuntimeInfo(ExecutorInfo executorInfo) {
        DataflowExecutorInfoVo executorInfoVo = new DataflowExecutorInfoVo();
        BeanUtil.copyProperties(executorInfo, executorInfoVo);
        // 通过uuid来删除此执行器
        priorityQueue.remove(executorInfoVo, Comparator.comparing(DataflowExecutorInfo::getUuid));
    }

    private void updateExecutorRuntimeInfo(ExecutorInfo executorInfo) {
        DataflowExecutorInfoVo executorInfoVo = new DataflowExecutorInfoVo();
        BeanUtil.copyProperties(executorInfo, executorInfoVo);
        try {
            // 请求执行器提供的接口获取运行时信息
            executorInfoService.setExecutorRuntimeInfo(executorInfoVo);
            // 将此执行器放入优先队列
            priorityQueue.enqueue(executorInfoVo, Comparator.comparing(DataflowExecutorInfoVo::getUuid));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 通过请求执行器，获取其任务列表，加载到executorInfo中
     *
     * @param executorInfo 执行器信息
     */
    private void reloadTaskListByNetwork(ExecutorInfo executorInfo) {
        try {
            String allTaskUrl = executorInfo.getExecutorUrl() + "/task/running";
            try (HttpResponse resp = HttpUtil.createGet(allTaskUrl)
                    .setConnectionTimeout(8000)
                    .execute()) {
                String body = resp.body();
                JSONObject json = ResponseUtil.toJson(body);
                if (ResponseUtil.isSuccess(json)) {
                    JSONArray jsonArray = json.getJSONObject("data").getJSONArray("data");
                    List<TaskInfo> taskInfos = new ArrayList<>();
                    for (Object o : jsonArray) {
                        TaskInfo taskInfo = JSONUtil.toBean((JSONObject) o, TaskInfo.class);
                        taskInfos.add(taskInfo);
                    }
                    executorInfo.setTaskList(taskInfos);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private String getPingUrl(ExecutorInfo executorInfo) {
        return "http://" + executorInfo.getHost() + ":" + executorInfo.getPort() + "/ping";
    }

    public ExecutorInfo getExecutorInfoByUuid(String uuid) {
        synchronized (execInfoMap) {
            return execInfoMap.get(uuid);
        }
    }

    /**
     * 向执行器添加任务
     *
     * @param taskInfo 任务
     */
    public void addTaskToExecutor(TaskInfo taskInfo) {
        ExecutorInfo executorInfo = taskInfo.getExecutorInfo();
        synchronized (execInfoMap) {
            ExecutorInfo executor = execInfoMap.get(executorInfo.getUuid());
            for (TaskInfo info : executor.getTaskList()) {
                if (StrUtil.equals(info.getId(), taskInfo.getId())) {
                    return;
                }
            }
            executor.getTaskList().add(taskInfo);
        }
    }

    /**
     * 在执行器上线前阻塞所有调用此方法的线程
     *
     * @throws InterruptedException 被打断时抛出
     */
    public void awaitUntilExecutorOnline() throws InterruptedException {
        if (CollUtil.isNotEmpty(this.getOnlineExecutors())) {
            return;
        }
        EXECUTOR_OFFLINE_LOCK.lock();
        try {
            if (CollUtil.isNotEmpty(this.getOnlineExecutors())) {
                return;
            }
            EXECUTOR_OFFLINE_CONDITION.await();
        } finally {
            EXECUTOR_OFFLINE_LOCK.unlock();
        }
    }
}
